package voldemort.store.cachestore;

import java.util.Properties;
import java.io.InputStream;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.SortedMap;
import java.util.Map;
import java.util.HashMap;
import voldemort.client.SocketStoreClientFactory;
import voldemort.client.ClientConfig;
import voldemort.versioning.Versioned;
import java.util.Random;
import voldemort.client.StoreClient;
import static voldemort.store.cachestore.BlockUtil.longToBytes;

/**
 * Created by IntelliJ IDEA.
 * User: mhsieh
 * Date: 2/14/11
 * Time: 2:28 PM
 * To change this template use File | Settings | File Templates.
 */
public class RunPacing extends TestClient {

    public RunPacing(long keyRange, int size, Random random, StoreClient<byte[], byte[]> client, long times, long lbase,
            int threads, String[] tests, long pace, boolean show) {

        super( keyRange, size, random, client, times, lbase, threads, tests, pace, show);
    }


    public void delete() {

        int error = 0;
        long begin = System.currentTimeMillis();
        for ( long i=lbase ; i < lbase+times ; i++) {
            long b = System.nanoTime();
            // add lbase back to random number
            long key = i ;
            try {
                Versioned<byte[]> value = client.get(longToBytes(key) )  ;
                if ( value != null) {
                    // delete 5 % of data
                    //if ( key % 20 == 19)
                    client.delete( longToBytes(key) );
                    if ( client.get( longToBytes(key)) != null ) {
                        logger.error("fail to delete key "+i);
                        error ++;
                    }
                }
                else
                    logger.warn("can not find key "+key);

                long d = (System.nanoTime() - b)/ 1000;
                if ( pace > 0 && pace > d /1000 ) {
                    try {
                        Thread.sleep( (pace -d /1000));
                        // swallow exception
                    } catch (InterruptedException iex) {}
                }
            } catch (Exception ex) {
                logger.error( ex.getMessage());
            }
        }
        output(begin, System.currentTimeMillis());
        logger.info("error "+ error +" delete operation "+ times );
    }

    public void insert() {

        int error = 0;
        long begin = System.currentTimeMillis();
        for ( long i= lbase ; i < lbase+times ; i++) {
            long b = System.nanoTime();
            // add lbase back to random number
            long key = i ;
            try {
                client.put( longToBytes(key), new byte[512] );
                if ( client.get (longToBytes( key) ) == null ) {
                    logger.error(" fail to insert key "+key );
                    error ++;
                }
                long d = (System.nanoTime() - b)/ 1000;
                if ( pace > 0 && pace > d /1000 ) {
                    try {
                        Thread.sleep( (pace -d /1000));
                        // swallow exception
                    } catch (InterruptedException iex) {}
                }
            } catch (Exception ex) {
                logger.error( ex.getMessage());
            }
        }
        output(begin, System.currentTimeMillis());
        logger.info("error "+error +" insert operation "+times );
    }


    @Override
    public void randomRW() {
        // set up duration array
        long[] durations = new long[ (int) times];

        long begin = System.currentTimeMillis();
        for ( long i=0; i < times ; i++) {
            long b = System.nanoTime();
            // add lbase back to random number
            long key = genLong( keyRange -lbase) +lbase ;
            try {
                Versioned<byte[]> value = client.get(longToBytes(key) )  ;
                if ( value != null)
                    client.put(longToBytes(key), value);
                else
                    logger.warn("can not find key "+key);

                long d = (System.nanoTime() - b)/ 1000;
                durations[(int) i] = d;
                if ( pace > 0 && pace > d /1000 ) {
                    try {
                        Thread.sleep( (pace -d /1000));
                        // swallow exception
                    } catch (InterruptedException iex) {}
                }
            } catch (Exception ex) {
                durations[ (int) i] = -1;
                logger.error( ex.getMessage());
            }
        }
        output(begin, System.currentTimeMillis());
        processResult( durations);
    }

    public void processResult(long[] result){
        logger.info("Process results ....");
        SortedMap<Long, Integer> maps = new TreeMap<Long, Integer>();
        int skip = 0;
        long total = 0;
        int scale = 10 ;
        for ( int i =0 ; i < result.length ; i++) {
            if ( result[i] == -1 ) skip++;
            else {
                // reduce order of 1 unit measure of 50 micro seconds
                long x = result[i] / scale;
                if ( maps.containsKey(x) ) {
                    int n = maps.get( x ) + 1;
                    maps.put( x, n);
                }
                else maps.put(x, 1);
                total += result[i];
            }
        }
        long avg = total / ( result.length - skip) ;
        long dev = 0;
        for ( int i =0 ; i < result.length ; i++) {
            if ( result[i] != -1) {
                dev += ( result[i] - avg)*( result[i] -avg) ;
            }
        }
        int std = (int) Math.sqrt( (double) (dev / (result.length - skip)) );
        logger.info("Average time in micro seconds " + avg+" standard deviation "+std );
        // print out detail distribution for detail
        if (show ) {
            logger.info("Output the distribution key's scale of "+scale+" micro seconds");
            logger.info("Time   Frequency");
            Iterator<Long> it = maps.keySet().iterator();
            while ( it.hasNext() ) {
                long key = it.next();
                logger.info( key +"     "+maps.get(key));
            }
        }
    }

    @Override
    public void runTest(String[] actions) {
        for(String action : actions) {
            if ( action.equals("randomRW")) randomRW();
            else if (action.equals("randomRead")) randomRead();
            else if (action.equals("randomWrite")) randomWrite();
            else if (action.equals("seqRW")) seqRW();
            else if (action.equals("seqRead")) seqRead();
            else if (action.equals("seqWrite")) seqWrite();
            else if (action.equals("populate")) populate();
            else if (action.equals("populateFixSize")) populateFixSize();
            else if (action.equals("testFunction")) testFunction();
            else if (action.equals("insert")) insert();
            else if (action.equals("delete")) delete();
            else {
                logger.warn("no action support "+action);
            }
        }
    }

    public static RunPacing createClient(String[] args){
       String[] opts = new String[] {"-url","-store","-keyRange","-size", "-testcase" , "-thread", "-times",
            "-begin" ,"-pace" , "-show" ,"-prop" };
        String[] defaults = new String[] {"tcp://localhost:6666","test","100", "1024","randomRW", "1", "0" ,
            "0" ,"0", "false", "addon/client.properties"};
        String[] paras = CLIUtils.getOpts( args, opts, defaults);
        String url = paras[0];
        String store = paras[1];
        // load client propeties if it does existing, using 10th position
        InputStream ip = TestClient.getClientProperties(paras[10]);
        ClientConfig config ;
        if ( ip != null ) {
            Properties prop = new Properties();
            try {
                prop.load(ip);
                //swallow exception
            } catch (Exception ex) {}
            config = new ClientConfig( prop);
        }
        else config = new ClientConfig();

        SocketStoreClientFactory factory = new SocketStoreClientFactory(config.setBootstrapUrls(url));
        long keyRange = Long.valueOf(paras[2]);
        int size = Integer.valueOf( paras[3]);
        StoreClient<byte[], byte[]> client = factory.getStoreClient( store);
        String[] tests = (paras[4].split(","));
        int threads = Integer.valueOf(paras[5]);
        long times = Long.valueOf( paras[6]);
        if ( times == 0 ) times = keyRange;
        long begin = Long.valueOf( paras[7]);
        long pace = Long.valueOf(paras[8]);
        boolean show = Boolean.valueOf(paras[9]);
        logger.info("using "+ cmdLine( opts, defaults));
        return new RunPacing( keyRange , size, new Random(keyRange), client , times, begin, threads, tests, pace, show ) ;
    }


    public static void main(String[] args) {
        RunPacing client = createClient( args);
//        if ( ! client.tests[0].equals("randomRW")) {
//            logger.warn("Pacing only support randomRW");
//            System.exit(1);
//        }
        if ( client.threads <= 1)
        // create a client that executes operations on a single store
            client.runTest( client.tests );
        else {
            if ( checkThread( client.tests)  ) {
                for ( int i=0; i < client.threads ; i++) {
                    new Thread(new RunThread( client, client.tests), "thread-"+i).start();
                }
            }
        }
    }
}
